package team.bluepen.supermarket.data.hbase;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.HTable;
import team.bluepen.supermarket.constant.Error;
import team.bluepen.supermarket.data.entity.MarketTable;
import team.bluepen.supermarket.util.Log;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 对HBase中的表进行操作。
 * @author Kai
 */
public class HTableRunner implements Closeable {
    private final MarketTable table;
    private final Admin admin;
    private final Connection connection;
    private final TableName name;
    private static final String TAG = "HTableRunner";

    public HTableRunner(MarketTable table) throws IOException {
        this.table = table;
        connection = HbaseConnectHelper.create();
        admin = connection.getAdmin();
        name = TableName.valueOf(getTableName());
    }

    public HTableRunner(HTable table) throws IOException {
        connection = HbaseConnectHelper.create();
        admin = connection.getAdmin();
        this.name = table.getName();
        this.table = new MarketTable();
        this.table.setName(name.getNameAsString());
        List<String> list = new ArrayList<>();
        HTableDescriptor hTableDescriptor = table.getTableDescriptor();
        for(HColumnDescriptor descriptor : hTableDescriptor.getColumnFamilies()){
            list.add(descriptor.getNameAsString());
        }
        this.table.setFamilies(list);
    }

    public String getTableName() {
        return table.getName();
    }

    public List<String> getTableFamilies() {
        return table.getFamilies();
    }

    /**
     * 创建表
     * @return {@link Error}
     */
    public Error create(){
        try {
            if (admin.tableExists(name)){
                return Error.ERROR_TABLE_EXISTED;
            }
            HTableDescriptor descriptor = new HTableDescriptor(name);

            for (String f: getTableFamilies()) {
                HColumnDescriptor cf = new HColumnDescriptor(f);
                descriptor.addFamily(cf);
            }

            admin.createTable(descriptor);
        } catch (IOException e) {
            Log.e(TAG, e);
            return Error.ERROR_DATABASE;
        }
        return Error.SUCCESS;
    }

    public Error addColumn(String column) {
        try {
            admin.addColumn(name, new HColumnDescriptor(column));
            List<String> l = new ArrayList<>(table.getFamilies());
            l.add(column);
            table.setFamilies(l);
            return Error.SUCCESS;
        } catch (IOException e){
            Log.e(TAG, e);
            return Error.ERROR_DATABASE;
        }
    }

    /**
     * 删除表
     * @return {@link Error}
     */
    public Error delete() {
        try {
            if (!admin.tableExists(name)){
                return Error.ERROR_TABLE_NOT_EXIST;
            }
            if (!admin.isTableDisabled(name)) {
                disable();
            }
            admin.deleteTable(name);
        } catch (IOException e) {
            Log.e(TAG, e);
            return Error.ERROR_DATABASE;
        }
        return Error.SUCCESS;
    }

    /**
     * 清空表
     * @return {@link Error}
     */
    public Error truncate() {
        try{
            if (!admin.isTableDisabled(name)) {
                disable();
            }
            admin.truncateTable(name, true);
            return Error.SUCCESS;
        } catch (IOException e){
            Log.e(TAG, e);
            return Error.ERROR_DATABASE;
        }
    }

    /**
     * 禁用表
     * @return {@link Error}
     */
    public Error disable() {
        try{
            admin.disableTableAsync(name);
            return Error.SUCCESS;
        } catch (IOException e){
            Log.e(TAG, e);
            return Error.ERROR_DATABASE;
        }
    }

    /**
     * 启用表
     * @return {@link Error}
     */
    public Error enable() {
        try{
            admin.enableTable(name);
            return Error.SUCCESS;
        } catch (IOException e){
            Log.e(TAG, e);
            return Error.ERROR_DATABASE;
        }
    }

    public void close() throws IOException {
        admin.close();
    }
}
